package org.apache.ocean.udp;

import java.io.*;
import java.net.*;
import java.util.*;
import java.util.concurrent.*;
import org.apache.ocean.*;
import java.util.logging.*;
import org.apache.commons.lang.*;
import org.apache.ocean.HostPort;


/**
 * Communicates via UDP.  Discovers other endpoints through the name services.
 *
 * @author jasonr
 */
public class UDPCommunication2 {
  public static Logger log = Logger.getLogger(UDPCommunication2.class.getName());
  public static final int BUFFER_SIZE = 4*1024;
  public DatagramSocket udpSocket;
  public ThreadPoolExecutor listenerThreadPool;
  public ListenerTask listenerTask;
  public int udpPort;
  public String thisID;
  public String thisRole;
  public List<Callback<UDPMessage,Boolean>> listeners = new ArrayList<Callback<UDPMessage,Boolean>>();
  
  public UDPCommunication2() {
  }
  
  protected void init(String thisID, int udpPort) throws Exception {
    this.thisID = thisID;
    this.udpPort = udpPort;
    udpSocket = new DatagramSocket(udpPort);
    listenerTask = new ListenerTask(udpSocket);
    listenerThreadPool = (ThreadPoolExecutor)Executors.newFixedThreadPool(1);
    listenerTask.future = listenerThreadPool.submit(listenerTask);
  }
  
  public void send(Set<HostPort> destinationHostPorts, UDPMessage udpMessage) throws Exception {
    for (HostPort hostPort : destinationHostPorts) {
      send(hostPort, udpMessage);
    }
  }
  
  public void send(HostPort hostPort, UDPMessage message) throws Exception {
    message.fromID = thisID;
    InetSocketAddress address = new InetSocketAddress(hostPort.host, hostPort.port);
    DatagramPacket packet = message.toPacket(address);
    udpSocket.send(packet);
  }
  
  private void fireMessageEvent(UDPMessage udpMessage) throws Exception {
    for (Callback<UDPMessage,Boolean> callback : listeners) {
      Boolean hasHandled = callback.call(udpMessage);
      if (BooleanUtils.isTrue(hasHandled)) return;
    }
    handleMessage(udpMessage);
  }
  
  protected void handleMessage(UDPMessage udpMessage) throws Exception {
    
  }

  public class ListenerTask implements Runnable {
    public Future future;
    public boolean doRun = true;
    DatagramSocket datagramSocket;
    
    public ListenerTask(DatagramSocket datagramSocket) {
      this.datagramSocket = datagramSocket;
    }
    
    public void run() {
      try {
        while (doRun) {
          DatagramPacket datagramPacket = new DatagramPacket(new byte[BUFFER_SIZE], BUFFER_SIZE);
          datagramSocket.receive(datagramPacket);
          String string = UDPUtil.toString(datagramPacket);
          UDPMessage udpMessage = new UDPMessage(string);
          if (!StringUtils.equals(thisID, udpMessage.fromID)) {
            handleMessage(udpMessage);
          }
        }
      } catch (Throwable throwable) {
        log.log(Level.SEVERE, "", throwable);
      }
    }
    
    public void stop() {
      doRun = false;
      if (future != null) future.cancel(true);
    }
  }
  
  public void close() {
    if (listenerTask != null) listenerTask.stop();
    if (listenerThreadPool != null) listenerThreadPool.shutdown();
    if (udpSocket != null) udpSocket.close();
  }
  
  public UDPMessage createUDPMessage(String action) throws Exception {
    UDPMessage message = new UDPMessage();
    message.fromRole = thisRole;
    message.action = action;
    message.fromID = thisID;
    return message;
  }
}
